package app_kvServer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Map;

import org.apache.log4j.Logger;

import common.messages.KVMessage;
import common.messages.KVMessageImpl;

/**
 * @author Zhou, Shi, Huang
 * 
 */
public class ClientConnection implements Runnable {
    private static Logger logger = Logger.getRootLogger();

    private boolean isOpen;
    private static final int BUFFER_SIZE = 1024;
    private static final int DROP_SIZE = 128 * BUFFER_SIZE;
    private static final int KEY_LENGTH = 20;
    private static final int VALUE_LENGTH = 120000;

    private Socket clientSocket;
    private InputStream input;
    private OutputStream output;
    private Map<String, String> kvmap;

    /**
     * @param client
     *            an client socket connection
     * @param map
     *            a hash map
     */
    public ClientConnection(Socket client, Map<String, String> map) {
        this.clientSocket = client;
        this.isOpen = true;
        this.kvmap = map;
    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        try {
            output = clientSocket.getOutputStream();
            input = clientSocket.getInputStream();

            // sendMessage(new
            // KVMessageImpl("Connection to MSRG Echo server established: "
            // +clientSocket.getLocalAddress().toString()+" "
            // + clientSocket.getLocalPort(),"" ,"GET_ERROR"));

            while (isOpen) {
                try {
                    KVMessageImpl latestMsg = receiveMessage();
                    if (latestMsg != null) {
                        if (latestMsg.getStatus().toString().contains("GET")) {
                            String clientValue = getKV(latestMsg.getKey());
                            if (clientValue.isEmpty()) {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), clientValue,
                                        "GET_ERROR"));
                            } else {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), clientValue,
                                        "GET_SUCCESS"));
                            }
                        } else if (latestMsg.getValue().toString()
                                .equalsIgnoreCase("null")) {
                            if (deleteKV(latestMsg.getKey())) {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(),
                                        latestMsg.getValue(), "DELETE_SUCCESS"));
                            } else {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(),
                                        latestMsg.getValue(), "DELETE_ERROR"));
                            }
                        } else if (kvmap.containsKey(latestMsg.getKey())) {
                            if (updateKV(latestMsg.getKey(),
                                    latestMsg.getValue())) {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), kvmap.get(latestMsg
                                                .getKey()), "PUT_UPDATE"));
                            } else {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), kvmap.get(latestMsg
                                                .getKey()), "PUT_ERROR"));
                            }
                        } else {
                            if (insertKV(latestMsg.getKey(),
                                    latestMsg.getValue())) {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), kvmap.get(latestMsg
                                                .getKey()), "PUT_SUCCESS"));
                            } else {
                                sendMessage(new KVMessageImpl(
                                        latestMsg.getKey(), kvmap.get(latestMsg
                                                .getKey()), "PUT_ERROR"));
                            }
                        }

                    }
                } catch (IOException ioe) {
                    logger.error("Error! Connection lost!");
                    isOpen = false;
                }
            }

        } catch (IOException ioe) {
            logger.error("Error! Connection could not be established!", ioe);

        } finally {
            try {
                if (clientSocket != null) {
                    input.close();
                    output.close();
                    clientSocket.close();
                }
            } catch (IOException ioe) {
                logger.error("Error! Unable to tear down connection!", ioe);
            }
        }
    }

    /**
     * @param msg
     * @throws IOException
     */
    public synchronized void sendMessage(KVMessage msg) throws IOException {
        KVMessageImpl kvm = new KVMessageImpl(msg.getKey(), msg.getValue(), msg
                .getStatus().toString());
        byte[] msgBytes = kvm.toByteArray(kvm);
        output.write(msgBytes, 0, msgBytes.length);
        output.flush();
        // logger.info("SEND \t<"
        // + clientSocket.getInetAddress().getHostAddress() + ":"
        // + clientSocket.getPort() + ">: '"
        // + msg.getMsg() +"'");
    }

    /**
     * @return
     * @throws IOException
     */
    private synchronized KVMessageImpl receiveMessage() throws IOException {

        int index = 0;
        byte[] msgBytes = null, tmp = null;
        byte[] bufferBytes = new byte[BUFFER_SIZE];

        /* read first char from stream */
        byte read = (byte) input.read();
        boolean reading = true;

        while (read != 13 && reading) {/* carriage return */
            /* if buffer filled, copy to msg array */
            if (index == BUFFER_SIZE) {
                if (msgBytes == null) {
                    tmp = new byte[BUFFER_SIZE];
                    System.arraycopy(bufferBytes, 0, tmp, 0, BUFFER_SIZE);
                } else {
                    tmp = new byte[msgBytes.length + BUFFER_SIZE];
                    System.arraycopy(msgBytes, 0, tmp, 0, msgBytes.length);
                    System.arraycopy(bufferBytes, 0, tmp, msgBytes.length,
                            BUFFER_SIZE);
                }

                msgBytes = tmp;
                bufferBytes = new byte[BUFFER_SIZE];
                index = 0;
            }

            /* only read valid characters, i.e. letters and constants */
            bufferBytes[index] = read;
            index++;

            /* stop reading is DROP_SIZE is reached */
            if (msgBytes != null && msgBytes.length + index >= DROP_SIZE) {
                reading = false;
            }

            /* read next char from stream */
            read = (byte) input.read();
        }

        if (msgBytes == null) {
            tmp = new byte[index];
            System.arraycopy(bufferBytes, 0, tmp, 0, index);
        } else {
            tmp = new byte[msgBytes.length + index];
            System.arraycopy(msgBytes, 0, tmp, 0, msgBytes.length);
            System.arraycopy(bufferBytes, 0, tmp, msgBytes.length, index);
        }

        msgBytes = tmp;

        // logger.info("RECEIVE \t<"
        // + clientSocket.getInetAddress().getHostAddress() + ":"
        // + clientSocket.getPort() + ">: '"
        // + msg.getMsg().trim() + "'");
        return KVMessageImpl.toKVMessage(msgBytes);
    }

    /**
     * @param key
     *            the new key
     * @param value
     *            the new value
     * @return
     */
    private synchronized boolean insertKV(String key, String value) {
        synchronized (this) {
            if (key.length() > KEY_LENGTH || value.length() > VALUE_LENGTH) {
                return false;
            } else {
                kvmap.put(key, value);
                return true;
            }
        }
    }

    /**
     * @param key
     *            the key ready to be updated
     * @param value
     *            the new value
     * @return
     */
    private synchronized boolean updateKV(String key, String value) {
        synchronized (this) {
            if (key.length() > KEY_LENGTH || value.length() > VALUE_LENGTH) {
                return false;
            } else {
                kvmap.remove(key);
                kvmap.put(key, value);
                return true;
            }
        }
    }

    /**
     * @param key
     * @return success or not
     */
    private synchronized boolean deleteKV(String key) {
        synchronized (this) {
            if (kvmap.containsKey(key)) {
                kvmap.remove(key);
                return true;
            } else {
                return false;
            }
        }
    }

    /**
     * @param key
     * @return the key and value
     */
    private String getKV(String key) {
        synchronized (this) {
            if (kvmap.containsKey(key)) {
                return kvmap.get(key).toString();
            } else
                return "";
        }
    }
}
